package test1;


import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;

public class PushConsumerExample {

    private static final Logger logger = LoggerFactory.getLogger(PushConsumerExample.class);


    private  PushConsumerExample() {

    }

    public static void main(String[] args) throws ClientException, InterruptedException {
        final ClientServiceProvider provider = ClientServiceProvider.loadService();

        //接入点地址
        String endpoint = "127.0.0.1:9876";
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
                .setEndpoints(endpoint)
                .build();

        //订阅消息的过滤规则
        String tag = "*";//匹配所有规则
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        //为消费者指定所属的消费者分组，Group需要提前创建
        String consumerGroup = "TestConsumerGroup";
        //指定需要订阅的目标topic
        String topic = "TestTopic";
        //初始化PushConsumer，需要绑定消费者分组,通信参数和订阅者关系
        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                .setConsumerGroup(consumerGroup)
                .setSubscriptionExpressions(Collections.singletonMap(topic,filterExpression))
                .setMessageListener(messageView -> {
                    //处理消息并返回消费结果
                    logger.info("Consume message successfully, messageiId={}",messageView.getMessageId());
                    return ConsumeResult.SUCCESS;
                })
                .build();
        Thread.sleep(Long.MAX_VALUE);

    }
}
